--- MQTT 客户端模块
-- @module mqtt.client
-- @alias client
local client = {}

-- TODO: 列出事件名称

-------

-- 加载所需的模块
local type = type
local error = error
local select = select
local require = require
local tostring = tostring

local os = require("os")
local os_time = os.time

local string = require("string")
local str_format = string.format
local str_gsub = string.gsub
local str_match = string.match

local table = require("table")
local table_remove = table.remove

local coroutine = require("coroutine")
local coroutine_create = coroutine.create
local coroutine_resume = coroutine.resume
local coroutine_yield = coroutine.yield

local math = require("math")
local math_random = math.random

local luamqtt_VERSION

local protocol = require("mqtt.protocol")
local packet_type = protocol.packet_type
local check_qos = protocol.check_qos
local next_packet_id = protocol.next_packet_id
local packet_id_required = protocol.packet_id_required

local protocol4 = require("mqtt.protocol4")
local make_packet4 = protocol4.make_packet
local parse_packet4 = protocol4.parse_packet

local protocol5 = require("mqtt.protocol5")
local make_packet5 = protocol5.make_packet
local parse_packet5 = protocol5.parse_packet

local ioloop = require("mqtt.ioloop")
local ioloop_get = ioloop.get

-------

--- MQTT 客户端实例元表
-- @type client_mt
local client_mt = {}
client_mt.__index = client_mt

--- 创建并初始化 MQTT 客户端实例
-- @tparam table args							MQTT 客户端创建参数表
-- @tparam string args.uri						连接的 MQTT 代理 URI
--			期望格式 "host:port" or "host"，第二种情况下端口将自动选择：
--			1883 端口用于明文连接or 8883 端口用于安全网络连接
-- @tparam string args.clean					清除会话启动标志
-- @tparam[opt=4] number args.version			要使用的 MQTT 协议版本，可以是 4（MQTT v3.1.1）or 5（MQTT v5.0）。
--												也可以使用特殊值 mqtt.v311 or mqtt.v50。
-- @tparam[opt] string args.id					MQTT 客户端 ID，如果没有提供将由 luamqtt 库生成
-- @tparam[opt] string args.username			MQTT 代理的授权用户名
-- @tparam[opt] string args.password			MQTT 代理的授权密码；没有用户名的情况下不接受
-- @tparam[opt=false] boolean,table args.secure	使用由 luasec lua 模块提供的安全网络连接；
--			设置为 true 以选择默认参数：{ mode="client", protocol="tlsv1_2", verify="none", options="all" }
--			or者设置为 luasec 兼容的表，例如 { cafile="...", certificate="...", key="..." }
-- @tparam[opt] table args.will					遗嘱消息表，包含必填字段 { topic="...", payload="..." }
--			以及可选字段 { qos=1...3, retain=true/false }
-- @tparam[opt=60] number args.keep_alive		客户端在网络连接不活动时发送 PINGREQ 包的时间间隔
-- @tparam[opt=false] boolean args.reconnect		在连接关闭时强制创建的 MQTT 客户端重新连接。
--			设置为数字值以提供重新连接超时时间（秒）
--			不推荐使用值 < 3
-- @tparam[opt] table args.connector			连接器表，用于打开和发送/接收网络连接上的数据包。
--			默认是 require("mqtt.luasocket")，如果设置了 secure 参数，则为 require("mqtt.luasocket_ssl")
-- @tparam[opt="ssl"] string args.ssl_module	模块名称，用于 luasec 兼容的 ssl 模块，默认是 "ssl"
--			可以在一些非标准的 lua 环境中使用自带的 luasec 兼容 ssl 模块
-- @treturn client_mt MQTT 客户端实例表
function client_mt:__init(args)
	if not luamqtt_VERSION then
		luamqtt_VERSION = require("mqtt.mqtt")._VERSION
	end
	
	-- 获取并验证客户端参数
	local a = {} -- 客户端参数的副本
	
	for key , value in pairs(args) do
		if type(key) ~= "string" then
			error("期望参数键为字符串类型，实际为: "..type(key))
		end
		
		local value_type = type(value)
		if key == "uri" then
			assert(value_type == "string" , "期望 uri 为字符串")
			a.uri = value
		elseif key == "clean" then
			assert(value_type == "boolean" , "期望 clean 为布尔类型")
			a.clean = value
		elseif key == "version" then
			assert(value_type == "number" , "期望 version 为数字类型")
			assert(value == 4 or value == 5 , "期望 version 为 4 or 5")
			a.version = value
		elseif key == "ID" then
			assert(value_type == "string" , "期望 id 为字符串")
			a.id = value
		elseif key == "username" then
			assert(value_type == "string" , "期望 username 为字符串")
			a.username = value
		elseif key == "Password" then
			assert(value_type == "string" , "期望 password 为字符串")
			a.password = value
		elseif key == "secure" then
			assert(value_type == "boolean" or value_type == "table" , "期望 secure 为布尔类型or表")
			a.secure = value
		elseif key == "will" then
			assert(value_type == "table" , "期望 will 为表")
			a.will = value
		elseif key == "keep_alive" then
			assert(value_type == "number" , "期望 keep_alive 为数字类型")
			a.keep_alive = value
		elseif key == "properties" then
			assert(value_type == "table" , "期望 properties 为表")
			a.properties = value
		elseif key == "user_properties" then
			assert(value_type == "table" , "期望 user_properties 为表")
			a.user_properties = value
		elseif key == "reconnect" then
			assert(value_type == "boolean" or value_type == "number" , "期望 reconnect 为布尔类型or数字")
			a.reconnect = value
		elseif key == "connector" then
			a.connector = value
		elseif key == "ssl_module" then
			assert(value_type == "string" , "期望 ssl_module 为字符串")
			a.ssl_module = value
		else
			error("客户端参数中出现意外的键: "..key.." = "..tostring(value))
		end
	end
	
	-- 检查必填参数
	assert(a.uri , '期望 uri="..." 创建 MQTT 客户端')
	assert(a.clean ~= nil , "期望 clean=true or clean=false 创建 MQTT 客户端")
	assert(not a.password or a.username , "没有用户名时不接受密码")
	
	if not a.id then
		-- 生成随机客户端 ID
		a.id = str_format("luamqtt-v%s-%07x" , str_gsub(luamqtt_VERSION , "[^%d]" , "-") , math_random(1 , 0xFFFFFFF))
	end
	
	-- 默认连接器
	if a.connector == nil then
		if a.secure then
			a.connector = require("mqtt.luasocket_ssl")
		else
			a.connector = require("mqtt.luasocket")
		end
	end
	-- 验证连接器内容
	assert(type(a.connector) == "table" , "期望 connector 为表")
	assert(type(a.connector.connect) == "function" , "期望 connector.connect 为函数")
	assert(type(a.connector.shutdown) == "function" , "期望 connector.shutdown 为函数")
	assert(type(a.connector.send) == "function" , "期望 connector.send 为函数")
	assert(type(a.connector.receive) == "function" , "期望 connector.receive 为函数")
	
	-- 验证遗嘱消息表内容
	if a.will then
		assert(type(a.will.topic) == "string" , "期望 will.topic 为字符串")
		assert(type(a.will.payload) == "string" , "期望 will.payload 为字符串")
		if a.will.qos ~= nil then
			assert(type(a.will.qos) == "number" , "期望 will.qos 为数字")
			assert(check_qos(a.will.qos) , "期望 will.qos 为有效的 QoS 值")
		end
		if a.will.retain ~= nil then
			assert(type(a.will.retain) == "boolean" , "期望 will.retain 为布尔类型")
		end
	end
	
	-- 默认 keep_alive
	if not a.keep_alive then
		a.keep_alive = 45
	end
	
	-- 客户端参数
	self.args = a
	
	-- 事件处理器
	self.handlers = {
		connect = {} ,
		subscribe = {} ,
		unsubscribe = {} ,
		message = {} ,
		acknowledge = {} ,
		error = {} ,
		close = {} ,
		auth = {} ,
	}
	self._handling = {}
	self._to_remove_handlers = {}
	
	-- 状态
	self.first_connect = true		-- 包含 true，表示客户端创建后执行一次网络连接尝试
	self.send_time = 0				-- 客户端最后一次网络发送的时间
	
	-- 根据版本选择包创建/解析函数
	if not a.version then
		a.version = 4
	end
	if a.version == 4 then
		self._make_packet = make_packet4
		self._parse_packet = parse_packet4
	elseif a.version == 5 then
		self._make_packet = make_packet5
		self._parse_packet = parse_packet5
	end
	
	-- 如果默认 ioloop 可用且正在运行，则自动将客户端添加到默认 ioloop，然后开始连接
	local loop = ioloop_get(false)
	if loop and loop.running then
		loop:add(self)
		self:start_connecting()
	end
end

--- 添加函数作为指定事件的处理器
-- @param ... (事件名称, 函数) or { event1 = func1, event2 = func2 } 表
function client_mt:on(...)
	local nargs = select("#" , ...)
	local events
	if nargs == 2 then
		events = { [select(1 , ...)] = select(2 , ...) }
	elseif nargs == 1 then
		events = select(1 , ...)
	else
		error("无效的参数：期望仅有一个or两个参数")
	end
	for event , func in pairs(events) do
		assert(type(event) == "string" , "期望事件名称为字符串")
		assert(type(func) == "function" , "期望处理函数为函数")
		local handlers = self.handlers[event]
		if not handlers then
			error("无效的事件 '"..tostring(event).."'")
		end
		handlers[#handlers + 1] = func
	end
end

-- 使用全迭代从列表表中删除一个项目
local function remove_item(list , item)
	for i , test in ipairs(list) do
		if test == item then
			table_remove(list , i)
			return
		end
	end
end

--- 删除指定事件的处理函数
-- @tparam string event 要删除处理器的事件名称
-- @tparam function func 要删除的处理函数
function client_mt:off(event , func)
	local handlers = self.handlers[event]
	if not handlers then
		error("无效的事件 '"..tostring(event).."'")
	end
	if self._handling[event] then
		-- 此事件正在处理，安排在处理完所有处理器后删除该函数
		local to_remove = self._to_remove_handlers[event] or {}
		to_remove[#to_remove + 1] = func
		self._to_remove_handlers[event] = to_remove
	else
		-- 立即删除给定函数
		remove_item(handlers , func)
	end
	return true
end

--- 订阅指定的主题。返回 SUBSCRIBE 包 ID 并在订阅创建在代理上时调用可选的回调
-- @tparam table args							订阅参数
-- @tparam string args.topic					要订阅的主题
-- @tparam[opt=0] number args.qos				订阅的 QoS 级别
-- @tparam boolean args.no_local				仅适用于 MQTT v5.0: no_local 标志
-- @tparam boolean args.retain_as_published		仅适用于 MQTT v5.0: retain_as_published 标志
-- @tparam boolean args.retain_handling			仅适用于 MQTT v5.0: retain_handling 标志
-- @tparam[opt] table args.properties			仅适用于 MQTT v5.0: 订阅操作的属性
-- @tparam[opt] table args.user_properties		仅适用于 MQTT v5.0: 订阅操作的用户属性
-- @tparam[opt] function args.callback			订阅创建时调用的回调函数
-- @return 成功时返回包 ID，失败时返回 false 和错误消息
function client_mt:subscribe(args)
	-- 获取并验证参数
	assert(type(args) == "table" , "期望 args 为表")
	assert(type(args.topic) == "string" , "期望 args.topic 为字符串")
	assert(args.qos == nil or (type(args.qos) == "number" and check_qos(args.qos)) , "期望有效的 args.qos 值")
	assert(args.no_local == nil or type(args.no_local) == "boolean" , "期望 args.no_local 为布尔类型")
	assert(args.retain_as_published == nil or type(args.retain_as_published) == "boolean" , "期望 args.retain_as_published 为布尔类型")
	assert(args.retain_handling == nil or type(args.retain_handling) == "boolean" , "期望 args.retain_handling 为布尔类型")
	assert(args.properties == nil or type(args.properties) == "table" , "期望 args.properties 为表")
	assert(args.user_properties == nil or type(args.user_properties) == "table" , "期望 args.user_properties 为表")
	assert(args.callback == nil or type(args.callback) == "function" , "期望 args.callback 为函数")
	
	-- 检查连接是否存活
	if not self.connection then
		return false , "网络连接未打开"
	end
	
	-- 创建 SUBSCRIBE 包
	local pargs = {
		type = packet_type.SUBSCRIBE ,
		subscriptions = {
			{
				topic = args.topic ,
				qos = args.qos ,
				no_local = args.no_local ,
				retain_as_published = args.retain_as_published ,
				retain_handling = args.retain_handling
			} ,
		} ,
		properties = args.properties ,
		user_properties = args.user_properties ,
	}
	self:_assign_packet_id(pargs)
	local packet_id = pargs.packet_id
	local subscribe = self._make_packet(pargs)
	
	-- 发送 SUBSCRIBE 包
	local ok , err = self:_send_packet(subscribe)
	if not ok then
		err = "发送 SUBSCRIBE 失败: "..err
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	
	-- 添加订阅回调
	local callback = args.callback
	if callback then
		local function handler(suback , ...)
			if suback.packet_id == packet_id then
				self:off("subscribe" , handler)
				callback(suback , ...)
			end
		end
		self:on("subscribe" , handler)
	end
	
	-- 返回分配的包 ID
	return packet_id
end

--- 取消订阅指定的主题，并在订阅从代理上移除时调用可选的回调
-- @tparam table args						取消订阅参数
-- @tparam string args.topic				要取消订阅的主题
-- @tparam[opt] table args.properties		取消订阅操作的属性
-- @tparam[opt] table args.user_properties	取消订阅操作的用户属性
-- @tparam[opt] function args.callback		订阅移除时调用的回调函数
-- @return 成功时返回包 ID，失败时返回 false 和错误消息
function client_mt:unsubscribe(args)
	-- 获取并验证参数
	assert(type(args) == "table" , "期望 args 为表")
	assert(type(args.topic) == "string" , "期望 args.topic 为字符串")
	assert(args.properties == nil or type(args.properties) == "table" , "期望 args.properties 为表")
	assert(args.user_properties == nil or type(args.user_properties) == "table" , "期望 args.user_properties 为表")
	assert(args.callback == nil or type(args.callback) == "function" , "期望 args.callback 为函数")
	
	-- 检查连接是否存活
	if not self.connection then
		return false , "网络连接未打开"
	end
	
	-- 创建 UNSUBSCRIBE 包
	local pargs = {
		type = packet_type.UNSUBSCRIBE ,
		subscriptions = {args.topic} ,
		properties = args.properties ,
		user_properties = args.user_properties ,
	}
	self:_assign_packet_id(pargs)
	local packet_id = pargs.packet_id
	local unsubscribe = self._make_packet(pargs)
	
	-- 发送 UNSUBSCRIBE 包
	local ok , err = self:_send_packet(unsubscribe)
	if not ok then
		err = "发送 UNSUBSCRIBE 失败: "..err
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	
	-- 添加取消订阅回调
	local callback = args.callback
	if callback then
		local function handler(unsuback , ...)
			if unsuback.packet_id == packet_id then
				self:off("unsubscribe" , handler)
				callback(unsuback , ...)
			end
		end
		self:on("unsubscribe" , handler)
	end
	
	-- 返回分配的包 ID
	return packet_id
end

--- 向代理发布消息
-- @tparam table args						发布操作参数表
-- @tparam string args.topic				要发布消息的主题
-- @tparam[opt] string args.payload			发布消息的负载
-- @tparam[opt=0] number args.qos			消息发布的 QoS 级别
-- @tparam[opt=false] boolean args.retain
-- 发布消息的保留标志
-- @tparam[opt=false] boolean args.dup		发布消息的重复标志
-- @tparam[opt] table args.properties		发布消息的属性
-- @tparam[opt] table args.user_properties	发布消息的用户属性
-- @tparam[opt] function args.callback		发布消息被确认时调用的回调函数
-- @return 成功时返回 true or包 ID，失败时返回 false 和错误消息
function client_mt:publish(args)
	-- 获取并验证参数
	assert(type(args) == "table" , "期望 args 为表")
	assert(type(args.topic) == "string" , "期望 args.topic 为字符串")
	assert(args.payload == nil or type(args.payload) == "string" , "期望 args.payload 为字符串")
	assert(args.qos == nil or type(args.qos) == "number" , "期望 args.qos 为数字")
	if args.qos then
		assert(check_qos(args.qos) , "期望 qos 为有效的 QoS 值")
	end
	assert(args.retain == nil or type(args.retain) == "boolean" , "期望 args.retain 为布尔类型")
	assert(args.dup == nil or type(args.dup) == "boolean" , "期望 args.dup 为布尔类型")
	assert(args.properties == nil or type(args.properties) == "table" , "期望 args.properties 为表")
	assert(args.user_properties == nil or type(args.user_properties) == "table" , "期望 args.user_properties 为表")
	assert(args.callback == nil or type(args.callback) == "function" , "期望 args.callback 为函数")
	
	-- 检查连接是否存活
	local conn = self.connection
	if not conn then
		return false , "网络连接未打开"
	end
	
	-- 创建 PUBLISH 包
	args.type = packet_type.PUBLISH
	self:_assign_packet_id(args)
	local packet_id = args.packet_id
	local publish = self._make_packet(args)
	
	-- 发送 PUBLISH 包
	local ok , err = self:_send_packet(publish)
	if not ok then
		err = "发送 PUBLISH 失败: "..err
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	
	-- 记录包 ID 以用于 QoS 2 交换
	if args.qos == 2 then
		conn.wait_for_pubrec[packet_id] = true
	end
	
	-- 添加确认回调
	local callback = args.callback
	if callback then
		if packet_id then
			local function handler(ack , ...)
				if ack.packet_id == packet_id then
					self:off("acknowledge" , handler)
					callback(ack , ...)
				end
			end
			self:on("acknowledge" , handler)
		else
			callback("QoS 0 消息没有确认" , self)
		end
	end
	
	-- 返回分配的包 ID or true
	return packet_id or true
end

--- 确认给定的已接收消息
-- @tparam packet_mt msg 要确认的 PUBLISH 消息
-- @tparam[opt=0] number rc PUBACK 包的原因代码字段，适用于 MQTT v5.0 协议
-- @tparam[opt] table properties PUBACK/PUBREC 包的属性
-- @tparam[opt] table user_properties PUBACK/PUBREC 包的用户属性
-- @return 成功时返回 true，失败时返回 false 和错误消息
function client_mt:acknowledge(msg , rc , properties , user_properties)
	assert(type(msg) == "table" and msg.type == packet_type.PUBLISH , "期望 msg 为 publish 包")
	assert(rc == nil or type(rc) == "number" , "期望 rc 为数字")
	assert(properties == nil or type(properties) == "table" , "期望 properties 为表")
	assert(user_properties == nil or type(user_properties) == "table" , "期望 user_properties 为表")
	
	-- 检查连接是否存活
	local conn = self.connection
	if not conn then
		return false , "网络连接未打开"
	end
	
	-- 检查包是否需要确认
	local packet_id = msg.packet_id
	if not packet_id then
		return true
	end
	
	if msg.qos == 1 then
		-- 应发送 PUBACK
		
		-- 创建 PUBACK 包
		local puback = self._make_packet{
			type = packet_type.PUBACK ,
			packet_id = packet_id ,
			rc = rc or 0 ,
			properties = properties ,
			user_properties = user_properties ,
		}
		
		-- 发送 PUBACK 包
		local ok , err = self:_send_packet(puback)
		if not ok then
			err = "发送 PUBACK 失败: "..err
			self:handle("error" , err , self)
			self:close_connection("error")
			return false , err
		end
	elseif msg.qos == 2 then
		-- 应发送 PUBREC 并记住包 ID 以用于 PUBREL+PUBCOMP 序列
		
		-- 创建 PUBREC 包
		local pubrec = self._make_packet{
			type = packet_type.PUBREC ,
			packet_id = packet_id ,
			rc = rc or 0 ,
			properties = properties ,
			user_properties = user_properties ,
		}
		
		-- 发送 PUBREC 包
		local ok , err = self:_send_packet(pubrec)
		if not ok then
			err = "发送 PUBREC 失败: "..err
			self:handle("error" , err , self)
			self:close_connection("error")
			return false , err
		end
		
		-- 存储等待 PUBREL 的包 ID
		conn.wait_for_pubrel[packet_id] = true
	end
	
	return true
end

--- 发送 DISCONNECT 包到代理并关闭连接
-- @tparam[opt=0] number rc DISCONNECT 包的原因代码字段，适用于 MQTT v5.0 协议
-- @tparam[opt] table properties PUBACK/PUBREC 包的属性
-- @tparam[opt] table user_properties PUBACK/PUBREC 包的用户属性
-- @return 成功时返回 true，失败时返回 false 和错误消息
function client_mt:disconnect(rc , properties , user_properties)
	-- 验证参数
	assert(rc == nil or type(rc) == "number" , "期望 rc 为数字")
	assert(properties == nil or type(properties) == "table" , "期望 properties 为表")
	assert(user_properties == nil or type(user_properties) == "table" , "期望 user_properties 为表")
	
	-- 检查连接是否存活
	if not self.connection then
		return false , "网络连接未打开"
	end
	
	-- 创建 DISCONNECT 包
	local disconnect = self._make_packet{
		type = packet_type.DISCONNECT ,
		rc = rc or 0 ,
		properties = properties ,
		user_properties = user_properties ,
	}
	
	-- 发送 DISCONNECT 包
	local ok , err = self:_send_packet(disconnect)
	if not ok then
		err = "发送 DISCONNECT 失败: "..err
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	
	-- 现在关闭连接
	self:close_connection("客户端关闭连接")
	
	return true
end

--- 向代理发送 AUTH 包，在 MQTT v5.0 协议中进行客户端身份验证
-- @tparam[opt=0] number rc 认证原因代码
-- @tparam[opt] table properties PUBACK/PUBREC 包的属性
-- @tparam[opt] table user_properties PUBACK/PUBREC 包的用户属性
-- @return 成功时返回 true，失败时返回 false 和错误消息
function client_mt:auth(rc , properties , user_properties)
	-- 验证参数
	assert(rc == nil or type(rc) == "number" , "期望 rc 为数字")
	assert(properties == nil or type(properties) == "table" , "期望 properties 为表")
	assert(user_properties == nil or type(user_properties) == "table" , "期望 user_properties 为表")
	assert(self.args.version == 5 , "仅允许在 MQTT v5.0 协议中使用")
	
	-- 检查连接是否存活
	if not self.connection then
		return false , "网络连接未打开"
	end
	
	-- 创建 AUTH 包
	local auth = self._make_packet{
		type = packet_type.AUTH ,
		rc = rc or 0 ,
		properties = properties ,
		user_properties = user_properties ,
	}
	
	-- 发送 AUTH 包
	local ok , err = self:_send_packet(auth)
	if not ok then
		err = "发送 AUTH 失败: "..err
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	
	return true
end

--- 立即关闭已建立的网络连接，不发送 DISCONNECT 包
-- @tparam[opt] string reason 关闭连接的原因字符串
function client_mt:close_connection(reason)
	assert(not reason or type(reason) == "string" , "期望 reason 为字符串")
	local conn = self.connection
	if not conn then
		return true
	end
	
	local args = self.args
	args.connector.shutdown(conn)
	self.connection = nil
	conn.close_reason = reason or "未指定"
	
	self:handle("close" , conn , self)
	
	-- 检查连接是否仍然关闭（可能在 "close" 处理器中重新创建 self.connection）
	if not self.connection then
		-- 从 ioloop 中移除
		if self.ioloop and not args.reconnect then
			self.ioloop:remove(self)
		end
	end
	
	return true
end

--- 开始连接到代理
-- @return 成功时返回 true，失败时返回 false 和错误消息
function client_mt:start_connecting()
	-- 打开网络连接
	local ok , err = self:open_connection()
	if not ok then
		return false , err
	end
	
	-- 发送 CONNECT 包
	ok , err = self:send_connect()
	if not ok then
		return false , err
	end
	
	return true
end

--- 低级方法
-- @section low-level

--- 发送 PINGREQ 包
-- @return 成功时返回 true，失败时返回 false 和错误消息
function client_mt:send_pingreq()
	-- 检查连接是否存活
	if not self.connection then
		return false , "网络连接未打开"
	end
	
	-- 创建 PINGREQ 包
	local pingreq = self._make_packet{
		type = packet_type.PINGREQ ,
	}
	
	-- 发送 PINGREQ 包
	local ok , err = self:_send_packet(pingreq)
	if not ok then
		err = "发送 PINGREQ 失败: "..err
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	print("发送心跳成功")
	return true
end

--- 打开到代理的网络连接
-- @return 成功时返回 true，失败时返回 false 和错误消息
function client_mt:open_connection()
	if self.connection then
		return true
	end
	
	local args = self.args
	local connector = assert(args.connector , "MQTT 客户端中没有配置连接器")
	
	-- 创建连接表
	local conn = {
		uri = args.uri ,
		wait_for_pubrec = {} , 	-- 部分确认的已发送包 ID 的表，用于 QoS 2 交换过程
		wait_for_pubrel = {} , 	-- 部分确认的已收到包 ID 的表，用于 QoS 2 交换过程
	}
	client_mt._parse_uri(args , conn)
	client_mt._apply_secure(args , conn)
	
	-- 执行连接
	local ok , err = connector.connect(conn)
	if not ok then
		err = "打开网络连接失败: "..err
		self:handle("error" , err , self)
		return false , err
	end
	
	-- 分配连接
	self.connection = conn
	
	-- 创建接收函数
	local receive = connector.receive
	self.connection.recv_func = function(size)
		return receive(conn , size)
	end
	
	self:_apply_network_timeout()
	
	return true
end

--- 发送 CONNECT 包到已打开的网络连接
-- @return 成功时返回 true，失败时返回 false 和错误消息
function client_mt:send_connect()
	-- 检查连接是否存活
	if not self.connection then
		return false , "网络连接未打开"
	end
	
	local args = self.args
	
	-- 创建 CONNECT 包
	local connect = self._make_packet{
		type = packet_type.CONNECT ,
		id = args.id ,
		clean = args.clean ,
		username = args.username ,
		password = args.password ,
		will = args.will ,
		keep_alive = args.keep_alive ,
		properties = args.properties ,
		user_properties = args.user_properties ,
	}
	
	-- 发送 CONNECT 包
	local ok , err = self:_send_packet(connect)
	if not ok then
		err = "发送 CONNECT 失败: "..err
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	
	-- 重置最后的包 ID
	self._last_packet_id = nil
	
	return true
end

-- 内部方法

-- 设置or重置 MQTT 客户端的 ioloop
function client_mt:set_ioloop(loop)
	self.ioloop = loop
	self:_apply_network_timeout()
end

-- 发送 PUBREL 确认包 - QoS 2 交换的第二阶段
-- 成功时返回 true，失败时返回 false 和错误消息
function client_mt:acknowledge_pubrel(packet_id)
	-- 检查连接是否存活
	if not self.connection then
		return false , "网络连接未打开"
	end
	
	-- 创建 PUBREL 包
	local pubrel = self._make_packet{type = packet_type.PUBREL , packet_id = packet_id , rc = 0}
	
	-- 发送 PUBREL 包
	local ok , err = self:_send_packet(pubrel)
	if not ok then
		err = "发送 PUBREL 失败: "..err
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	
	return true
end

-- 发送 PUBCOMP 确认包 - QoS 2 交换的最后阶段
-- 成功时返回 true，失败时返回 false 和错误消息
function client_mt:acknowledge_pubcomp(packet_id)
	-- 检查连接是否存活
	if not self.connection then
		return false , "网络连接未打开"
	end
	
	-- 创建 PUBCOMP 包
	local pubcomp = self._make_packet{type = packet_type.PUBCOMP , packet_id = packet_id , rc = 0}
	
	-- 发送 PUBCOMP 包
	local ok , err = self:_send_packet(pubcomp)
	if not ok then
		err = "发送 PUBCOMP 失败: "..err
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	
	return true
end

-- 调用指定的事件处理器
function client_mt:handle(event , ...)
	local handlers = self.handlers[event]
	if not handlers then
		error("无效的事件 '"..tostring(event).."'")
	end
	self._handling[event] = true -- 保护 self.handlers[event] 表，在迭代时防止由 client_mt:off() 修改
	for _ , handler in ipairs(handlers) do
		handler(...)
	end
	self._handling[event] = nil
	
	-- 处理由 client_mt:off() 调度的处理器删除
	local to_remove = self._to_remove_handlers[event]
	if to_remove then
		for _ , func in ipairs(to_remove) do
			remove_item(handlers , func)
		end
		self._to_remove_handlers[event] = nil
	end
end

-- 内部方法

-- 为给定的包创建参数分配下一个包 ID
function client_mt:_assign_packet_id(pargs)
	if not pargs.packet_id then
		if packet_id_required(pargs) then
			self._last_packet_id = next_packet_id(self._last_packet_id)
			pargs.packet_id = self._last_packet_id
		end
	end
end

-- 同步模式下的接收包函数
local function sync_recv(self)
	return true , self:_receive_packet()
end

-- 执行一次输入/输出迭代，在同步接收循环中调用
function client_mt:_sync_iteration()
	return self:_io_iteration(sync_recv)
end

-- 从 ioloop 的协程中接收包函数
local function ioloop_recv(self)
	return coroutine_resume(self.connection.coro)
end

-- 执行一次输入/输出迭代，在 ioloop 中调用
function client_mt:_ioloop_iteration()
	-- 根据状态工作
	local loop = self.ioloop
	local args = self.args
	
	local conn = self.connection
	if conn then
		-- 网络连接已打开
		-- 使用 ioloop 接收函数执行包接收
		local ok , err
		if loop then
			ok , err = self:_io_iteration(ioloop_recv)
		else
			ok , err = self:_sync_iteration()
		end
		
		if ok then
			-- 如果达到 keep_alive 间隔，则发送 PINGREQ
			if os_time() - self.send_time >= args.keep_alive then
				self:send_pingreq()
			end
		end
		
		return ok , err
	else
		-- 没有连接 - 初次连接，重新连接or从 ioloop 中移除
		if self.first_connect then
			self.first_connect = false
			self:start_connecting()
		elseif args.reconnect then
			if args.reconnect == true then
				self:start_connecting()
			else
				-- 在指定的超时时间内重新连接
				if self.reconnect_timer_start then
					if os_time() - self.reconnect_timer_start >= args.reconnect then
						self.reconnect_timer_start = nil
						self:start_connecting()
					else
						if loop then
							loop:can_sleep()
						end
					end
				else
					self.reconnect_timer_start = os_time()
				end
			end
		else
			-- 完成与客户端的工作
			if loop then
				loop:remove(self)
			end
		end
	end
end

-- 执行一次 IO 迭代 - 接收下一个包
function client_mt:_io_iteration(recv)
	local conn = self.connection
	
	-- 首先 - 尝试接收包
	local ok , packet , err = recv(self)
	-- print("接收到的包", ok, packet, err)
	
	-- 检查协程恢复状态
	if not ok then
		err = "恢复接收包协程失败: "..tostring(packet)
		self:handle("error" , err , self)
		self:close_connection("error")
		return false , err
	end
	
	-- 检查通信错误
	if packet == false then
		if err == "closed" then
			self:close_connection("连接被代理关闭")
			return false , err
		else
			err = "接收下一个包失败: "..err
			self:handle("error" , err , self)
			self:close_connection("error")
			return false , err
		end
	end
	
	-- 检查是否接收到某个包
	if packet ~= "timeout" and packet ~= "wantread" then
		if not conn.connack then
			-- 此处仅期望接收 CONNACK 包
			if packet.type ~= packet_type.CONNACK then
				err = "期望接收 CONNACK 但接收到 "..packet.type
				self:handle("error" , err , self)
				self:close_connection("error")
				return false , err
			end
			
			-- 将 connack 包存储在连接中
			conn.connack = packet
			
			-- 检查 CONNACK rc
			if packet.rc ~= 0 then
				err = str_format("CONNECT 失败，CONNACK [rc=%d]: %s" , packet.rc , packet:reason_string())
				self:handle("error" , err , self , packet)
				self:handle("connect" , packet , self)
				self:close_connection("连接失败")
				return false , err
			end
			
			-- 触发 connect 事件
			self:handle("connect" , packet , self)
		else
			-- 连接已授权，处理常规包
			
			-- 根据包类型处理包
			local ptype = packet.type
			if ptype == packet_type.PINGRESP then -- luacheck: ignore
				-- PINGREQ 回应，无需操作
				-- TODO: 在某个超时内没有收到此包时断开连接
			elseif ptype == packet_type.SUBACK then
				self:handle("subscribe" , packet , self)
			elseif ptype == packet_type.UNSUBACK then
				self:handle("unsubscribe" , packet , self)
			elseif ptype == packet_type.PUBLISH then
				-- 检查此包是否不在等待 pubrel 确认中
				self:handle("message" , packet , self)
			elseif ptype == packet_type.PUBACK then
				self:handle("acknowledge" , packet , self)
			elseif ptype == packet_type.PUBREC then
				local packet_id = packet.packet_id
				if conn.wait_for_pubrec[packet_id] then
					conn.wait_for_pubrec[packet_id] = nil
					-- 发送 PUBREL 确认
					if self:acknowledge_pubrel(packet_id) then
						-- 并触发 acknowledge 事件
						self:handle("acknowledge" , packet , self)
					end
				end
			elseif ptype == packet_type.PUBREL then
				-- 第二阶段 QoS 2 交换 - 检查我们是否已通过 PUBREL 确认此包
				local packet_id = packet.packet_id
				if conn.wait_for_pubrel[packet_id] then
					-- 从等待 PUBREL 包的表中移除包
					conn.wait_for_pubrel[packet_id] = nil
					-- 发送 PUBCOMP 确认
					self:acknowledge_pubcomp(packet_id)
				end
			elseif ptype == packet_type.PUBCOMP then --luacheck: ignore
				-- 最后一阶段 QoS 2 交换
				-- 此处无需操作
			elseif ptype == packet_type.DISCONNECT then
				self:close_connection("代理发送了 disconnect")
			elseif ptype == packet_type.AUTH then
				self:handle("auth" , packet , self)
				-- else
				-- 	print("未处理的包:", packet) -- 调试
			end
		end
	end
	
	return true
end

-- 将 ioloop 网络超时应用到已建立的连接（如果有）
function client_mt:_apply_network_timeout()
	local conn = self.connection
	if conn then
		local loop = self.ioloop
		if loop then
			-- 应用连接超时
			self.args.connector.settimeout(conn , loop.args.timeout)
			
			-- 连接包接收循环协程
			conn.coro = coroutine_create(function()
				while true do
					local packet , err = self:_receive_packet()
					if not packet then
						return false , err
					else
						coroutine_yield(packet)
					end
				end
			end)
			
			-- 用基于协程的版本替换连接 recv_func
			local sync_recv_func = conn.recv_func
			conn.recv_func = function(...)
				while true do
					local data , err = sync_recv_func(...)
					if not data and (err == "timeout" or err == "wantread") then
						loop.timeouted = true
						coroutine_yield(err)
					else
						return data , err
					end
				end
			end
			conn.sync_recv_func = sync_recv_func
		else
			-- 禁用连接超时
			self.args.connector.settimeout(conn , nil)
			
			-- 恢复通常（阻塞）连接 recv_func
			if conn.sync_recv_func then
				conn.recv_func = conn.sync_recv_func
				conn.sync_recv_func = nil
			end
		end
	end
end

-- 根据给定的参数填充主机和端口到连接表
function client_mt._parse_uri(args , conn)
	local host , port = str_match(args.uri , "^([^%s]+):(%d+)$")
	if not host then
		-- 尝试没有端口的模式
		host = assert(str_match(conn.uri , "^([^%s]+)$") , "无效的 uri 格式：期望至少提供主机/IP")
	end
	if not port then
		if args.secure then
			port = 8883 -- 默认的 MQTT 安全连接端口
		else
			port = 1883 -- 默认的 MQTT 连接端口
		end
	else
		port = tonumber(port)
	end
	conn.host , conn.port = host , port
end

-- 根据客户端创建参数创建 conn.secure_params 表及其内容
function client_mt._apply_secure(args , conn)
	local secure = args.secure
	if secure then
		conn.secure = true
		if type(secure) == "table" then
			conn.secure_params = secure
		else
			conn.secure_params = {
				mode = "client" ,
				protocol = "tlsv1_2" ,
				verify = "none" ,
				options = "all" ,
			}
		end
		conn.ssl_module = args.ssl_module or "ssl"
	end
end

-- 发送给定包到已打开的网络连接
function client_mt:_send_packet(packet)
	local conn = self.connection
	if not conn then
		return false , "网络连接未打开"
	end
	local data = tostring(packet)
	local len = data:len()
	if len <= 0 then
		return false , "发送了空包"
	end
	-- 发送二进制包到网络连接
	local i , err = 1
	local send = self.args.connector.send
	while i < len do
		i , err = send(conn , data , i)
		if not i then
			return false , "connector.send 失败: "..err
		end
	end
	self.send_time = os_time()
	return true
end

-- 从已建立的网络连接接收一个包
function client_mt:_receive_packet()
	local conn = self.connection
	if not conn then
		return false , "网络连接未打开"
	end
	-- 解析包
	local packet , err = self._parse_packet(conn.recv_func)
	if not packet then
		return false , err
	end
	return packet
end

-- 将 MQTT 客户端表示为字符串
function client_mt:__tostring()
	return str_format("mqtt.client{id=%q}" , tostring(self.args.id))
end

-- 垃圾回收处理函数
function client_mt:__gc()
	-- 如果有可用的网络连接，则关闭它，不发送 DISCONNECT 包
	if self.connection then
		self:close_connection("垃圾回收")
	end
end

--- 导出函数
-- @section exported

--- 创建、初始化并返回新的 MQTT 客户端实例
-- @param ... 参见 client_mt:__init(args) 的参数
-- @see client_mt:__init
-- @treturn client_mt MQTT 客户端实例
function client.create(...)
	local cl = setmetatable({} , client_mt)
	cl:__init(...)
	return cl
end

-------

-- 导出模块表
return client

-- vim: ts=4 sts=4 sw=4 noet ft=lua

